package com.emc.elc.rabbitmq.wapper;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import com.emc.elc.rabbitmq.wapper.Handler.MessageHandler;
import com.emc.elc.rabbitmq.wapper.exception.TopicExistedException;
import com.emc.elc.rabbitmq.wapper.exception.TopicNotExistedException;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ShutdownListener;

public class MessageQueueService {
    private Map<String, QueueConsumerThread> queueCosumerThreads = new HashMap<String, QueueConsumerThread>();
    private ConnectionFactory connectionfactory;
    private Connection connection;
    private Channel channel;
    private static MessageQueueService messageQueueService = null;
    private String exchange;

    public static boolean isMessageQueueServiceExist() {
        return null != messageQueueService;
    }

    public static MessageQueueService getExsitedInstance() {
        return messageQueueService;
    }

    public static MessageQueueService getInstance(String username, String password, String host, int port,
            String virtualHost, String exchange) throws IOException {
        if (null == messageQueueService) {
            messageQueueService = new MessageQueueService(username, password, host, port, virtualHost, exchange);
        }

        return messageQueueService;
    }

    private MessageQueueService(String username, String password, String host, int port, String virtualHost,
            String exchange) throws IOException {
        this.exchange = exchange;

        connectionfactory = new ConnectionFactory();
        connectionfactory.setUsername(username);
        connectionfactory.setPassword(password);
        connectionfactory.setHost(host);
        connectionfactory.setPort(port);
        connectionfactory.setVirtualHost(virtualHost);
        connection = connectionfactory.newConnection();
        channel = connection.createChannel();
    }

    public boolean sendMessage(String target, BasicProperties basicProperties, String message) throws IOException {
        boolean isSuccessful = false;
        this.channel.basicPublish(this.exchange, target, basicProperties, message.getBytes());
        System.out.println(" [x] Sent '" + target + "':'" + message + "'");
        isSuccessful = true;

        return isSuccessful;
    }
    
    public boolean bindQueue(String queue, MessageHandler queueHandler) throws TopicExistedException, IOException {
        if (queueCosumerThreads.containsKey(queue)) {
            throw new TopicExistedException("Cannot bind: Topic have existed, try add handler to this exception.");
        }
        
        QueueConsumerThread queueConsumerThread = new QueueConsumerThread(this.channel, queue, null, queueHandler, exchange);
        queueCosumerThreads.put(queue, queueConsumerThread);
        return true;
    }
    
    public boolean bindTopicToQueue(String topic, String queue, MessageHandler queueHandler) throws IOException, TopicExistedException {
        if (queueCosumerThreads.containsKey(topic)) {
            throw new TopicExistedException("Cannot bind: Topic have existed, try add handler to this exception.");
        }
        QueueConsumerThread queueConsumerThread = new QueueConsumerThread(this.channel, queue, topic, queueHandler, exchange);
        queueCosumerThreads.put(topic, queueConsumerThread);
        return true;
    }
    
    public String declareQueue (String queueId) throws IOException {
        this.channel.queueDeclare(queueId, false, false, false, null);
        return queueId;
    }
    
    public void deleteQueue (String queueId) throws IOException {
        this.channel.queueDelete(queueId);
    }

    public boolean addMessageHandler(final String topic, final MessageHandler queueHandler) throws TopicNotExistedException {
        if (!queueCosumerThreads.containsKey(topic)) {
            throw new TopicNotExistedException("Topic not existed. Try bind a topic to handler.");
        }

        new Thread(new Runnable() {
            @Override
            public void run() {
                QueueConsumerThread queueConsumerThread = queueCosumerThreads.get(topic);
                queueConsumerThread.stopListning();
                queueConsumerThread = queueConsumerThread.addMessageHandlers(queueHandler);
                queueConsumerThread.startListning();
            }
        }).start();

        return true;
    }
    
    public boolean removeMessageHandler(final String topic, final MessageHandler queueHandler) throws TopicNotExistedException {
        if (!queueCosumerThreads.containsKey(topic)) {
            throw new TopicNotExistedException("Topic not existed. Try bind a topic to handler.");
        }

        new Thread(new Runnable() {
            @Override
            public void run() {
                QueueConsumerThread queueConsumerThread = queueCosumerThreads.get(topic);
                queueConsumerThread.stopListning();
                queueConsumerThread = queueConsumerThread.removeMessageHandlers(queueHandler);
                queueConsumerThread.startListning();
            }
        }).start();

        return true;
    }

    public void removeTopic(String topic) throws IOException {
        if (queueCosumerThreads.containsKey(topic)) {
            queueCosumerThreads.get(topic).destroyConsumer();
            queueCosumerThreads.remove(topic);
        }
    }

    public void stopMessageQueueService() throws IOException {
        for (String queueName : queueCosumerThreads.keySet()) {
            removeTopic(queueName);
        }
        connection.close();
        messageQueueService = null;
    }

    public int getConnectionHeartBeat() {
        return this.connection.getHeartbeat();
    }

    public void addConnectionDownListener(ShutdownListener shutdownListner) {
        this.connection.addShutdownListener(shutdownListner);
    }

    public void finalize() throws Throwable {
        super.finalize();
        stopMessageQueueService();
    }
}
